package com.fyt.reachgo.consummer;

import com.fyt.reachgo.config.RabbitMQConfig;
import com.fyt.reachgo.domain.message.LikeMessage;
import com.fyt.reachgo.domain.UserLike;
import com.fyt.reachgo.mapper.TravelGuideMapper;
import com.fyt.reachgo.mapper.UserLikeMapper;
import com.rabbitmq.client.Channel;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.dao.DeadlockLoserDataAccessException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.sql.SQLTransientConnectionException;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


@Slf4j
@Component
@RequiredArgsConstructor
public class LikeConsumer {
    private final RedisTemplate<String, Object> redisTemplate;
    private final UserLikeMapper userLikeMapper;
    private final TravelGuideMapper travelGuideMapper;

    // 批量处理相关配置
    private static final int BATCH_SIZE = 10;
    private final List<LikeMessage> batchBuffer = Collections.synchronizedList(new ArrayList<>());
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    @PostConstruct
    public void init() {
        // 定时刷新缓冲区(每5秒检查一次)
        scheduler.scheduleAtFixedRate(this::flushBatch, 60, 60, TimeUnit.SECONDS);
    }

    @PreDestroy
    public void destroy() {
        scheduler.shutdown();
        // 应用关闭前强制刷新剩余消息
        flushBatch();
    }

    @RabbitListener(queues = RabbitMQConfig.LIKE_QUEUE, concurrency = "5-10")
    public void handleLikeMessage(LikeMessage message,
                                  Channel channel,
                                  @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.debug("接收点赞消息: {}", message);

            // 1. 先更新Redis缓存(快速响应)
            updateRedisCacheImmediately(message);

            // 2. 加入批量处理缓冲区
            synchronized (batchBuffer) {
                batchBuffer.add(message);
                if (batchBuffer.size() >= BATCH_SIZE) {
                    flushBatch();
                }
            }

            // 3. 立即确认消息(降低消息堆积风险)
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理点赞消息失败: {}", e.getMessage());
            handleMessageFailure(channel, deliveryTag, e);
        }
    }

    // 立即更新Redis缓存
    private void updateRedisCacheImmediately(LikeMessage message) {
        Long userId = message.getUserId();
        Integer guideId = message.getGuideId();

        String userLikeKey = String.format("like:user:%d:guide:%d", userId, guideId);
        String guideLikeCountKey = String.format("like:guide:%d:count", guideId);

        if (Boolean.TRUE.equals(message.getStatus())) {
            // 点赞操作
            System.out.println("点赞缓存操作开始执行");
            redisTemplate.opsForValue().set(userLikeKey, "1", 7, TimeUnit.DAYS);
            redisTemplate.opsForValue().increment(guideLikeCountKey);
        } else {
            // 取消点赞操作
            System.out.println("取消点赞缓存操作开始执行");
            redisTemplate.delete(userLikeKey);
            redisTemplate.opsForValue().decrement(guideLikeCountKey);
        }
    }

    // 批量刷新到数据库
    private void flushBatch() {
        List<LikeMessage> messagesToProcess;

        // 1. 安全获取待处理消息
        synchronized (batchBuffer) {
            if (batchBuffer.isEmpty()) {
                log.debug("【用户点赞队列】 ：批量刷新触发，但缓冲区为空，跳过处理");
                return;
            }
            messagesToProcess = new ArrayList<>(batchBuffer);
            batchBuffer.clear();
        }

        // 2. 空列表二次检查
        if (messagesToProcess.isEmpty()) {
            log.warn("获取到空消息列表，可能并发环境下的竞态条件，跳过处理");
            return;
        }

        log.info("开始批量处理 {} 条点赞消息", messagesToProcess.size());

        try {
            // 3. 按操作类型分组（空列表安全）
            Map<Boolean, List<LikeMessage>> groupedMessages = messagesToProcess.stream()
                    .collect(Collectors.partitioningBy(m -> Boolean.TRUE.equals(m.getStatus())));

            // 4. 处理前检查非空
            // 处理点赞
            processBatchGroup(groupedMessages.get(true), true);
            // 处理取消点赞
            processBatchGroup(groupedMessages.get(false), false);

            log.info("成功批量处理 {} 条点赞消息", messagesToProcess.size());
        } catch (Exception e) {
            log.error("批量处理点赞消息失败", e);
            handleFailedBatch(messagesToProcess, e);
        }
    }

    // 处理单个批处理组（点赞或取消点赞）
    private void processBatchGroup(List<LikeMessage> messages, boolean isLike) {
        log.debug("数据库开始前检查消息是否为空： {}  islike： {}" , messages,isLike);
        if (messages.isEmpty()) {
            log.debug("跳过处理{}操作，消息列表为空", isLike ? "点赞" : "取消点赞");
            return;
        }
        if (isLike) {
            batchProcessLikes(messages);
        } else {
            batchProcessUnlikes(messages);
        }
        log.debug("数据库操作正常结束");
    }

    // 处理批量失败的情况
    private void handleFailedBatch(List<LikeMessage> failedMessages, Exception e) {
        if (failedMessages.isEmpty()) {
            return;
        }

        // 根据异常类型决定是否重试
        if (shouldRetryBatch(e)) {
            log.warn("将 {} 条失败消息重新放回缓冲区等待重试", failedMessages.size());
            synchronized (batchBuffer) {
                batchBuffer.addAll(0, failedMessages); // 添加到队列头部优先处理
            }
        } else {
            log.error("放弃处理 {} 条消息，错误类型: {}", failedMessages.size(), e.getClass().getSimpleName());
            // 可以在这里添加死信队列发送或其他补偿逻辑
        }
    }

    // 判断批量操作是否应该重试
    private boolean shouldRetryBatch(Exception e) {
        // 数据库连接问题、死锁等可以重试
        if (e instanceof SQLTransientConnectionException ||
                e instanceof DeadlockLoserDataAccessException) {
            return true;
        }
        // 数据校验错误等不应重试
        if (e instanceof DataIntegrityViolationException) {
            return false;
        }
        // 默认重试
        return true;
    }

    // 批量处理点赞
    private void batchProcessLikes(List<LikeMessage> likeMessages) {
        log.debug("数据库方法被调用了：开始批量处理点赞");
        // 1. 过滤已存在的点赞(幂等性处理)
        List<LikeMessage> newLikes = likeMessages.stream()
                .filter(m -> !userLikeMapper.exists(m.getUserId(), m.getGuideId()))
                .collect(Collectors.toList());

        if (newLikes.isEmpty()) {
            return;
        }

        // 2. 批量插入点赞记录
        List<UserLike> userLikes = newLikes.stream()
                .map(m -> new UserLike(m.getUserId(), m.getGuideId(), new Date(m.getTimestamp())))
                .collect(Collectors.toList());
        userLikeMapper.batchInsert(userLikes);

        // 3. 批量更新路书点赞数(按guideId分组统计)
        Map<Integer, Long> guideLikeCounts = newLikes.stream()
                .collect(Collectors.groupingBy(LikeMessage::getGuideId, Collectors.counting()));

        guideLikeCounts.forEach((guideId, count) -> {
            travelGuideMapper.incrementLikes(guideId, count.intValue());
        });
    }

    // 批量处理取消点赞
    private void batchProcessUnlikes(List<LikeMessage> unlikeMessages) {
        // 1. 过滤不存在的点赞记录
        List<LikeMessage> validUnlikes = unlikeMessages.stream()
                .filter(m -> userLikeMapper.exists(m.getUserId(), m.getGuideId()))
                .collect(Collectors.toList());

        log.debug("批量处理取消点赞，validUnlikes： {}", validUnlikes);

        if (validUnlikes.isEmpty()) {
            return;
        }

        // 2. 批量删除点赞记录
        userLikeMapper.batchDelete(validUnlikes.stream()
                .map(m -> new UserLike(m.getUserId(), m.getGuideId(), null))
                .collect(Collectors.toList()));

        // 3. 批量更新路书点赞数(按guideId分组统计)
        Map<Integer, Long> guideUnlikeCounts = validUnlikes.stream()
                .collect(Collectors.groupingBy(LikeMessage::getGuideId, Collectors.counting()));

        guideUnlikeCounts.forEach((guideId, count) -> {
            travelGuideMapper.incrementLikes(guideId, -count.intValue());
        });
    }

    // 错误处理(与原版类似但更简单，因为大部分错误已在flushBatch中处理)
    private void handleMessageFailure(Channel channel, long deliveryTag, Exception e) {
        try {
            channel.basicNack(deliveryTag, false, false); // 不再重试，避免重复处理
            log.error("消息处理最终失败，已拒绝并不再重试: {}", deliveryTag);
        } catch (IOException ioException) {
            log.error("拒绝消息失败: {}", ioException.getMessage());
        }
    }
}

